package com.atguigu.gmall.realtime.app

import java.{lang, util}
import java.text.SimpleDateFormat
import java.time.{LocalDate, Period}
import java.util.Date

import com.alibaba.fastjson.{JSON, JSONObject}
import com.atguigu.gmall.realtime.bean.{DauInfo, PageLog}
import com.atguigu.gmall.realtime.util.{MyBeanUtils, MyEsUtil, MyKafkaUtil, OffsetManager, RedisUtil}
import org.apache.commons.lang3.ArrayUtils
import org.apache.commons.lang3.time.DateUtils
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{HasOffsetRanges, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.elasticsearch.ElasticsearchStatusException
import org.elasticsearch.action.search.{SearchRequest, SearchResponse}
import org.elasticsearch.client.{RequestOptions, RestHighLevelClient}
import org.elasticsearch.rest.RestStatus
import org.elasticsearch.search.SearchHit
import org.elasticsearch.search.builder.SearchSourceBuilder
import redis.clients.jedis.{Jedis, Pipeline}

import scala.collection.mutable.ListBuffer

object DwDauApp {


  //0 环境 ssc
  //1  读取偏移量
  //  2 接收kafka数据
  //3  提取偏移量结束点
  //4  转换结构 msg->   PageLog
  // 5 筛选去重
  // 6 合并维度
  //  7写入es
  //8  提交偏移量
  def main(args: Array[String]): Unit = {
    //0 环境 ssc
    val sparkConf: SparkConf = new SparkConf().setAppName("dw_dau_app").setMaster("local[4]")
    val ssc = new StreamingContext(sparkConf,Seconds(5))

    val topic="DWD_PAGE_LOG"
    val  groupId="dw_dau_group"

      //  0.5  加载数据前 还原状态
    revertDauState()

    //1  读取偏移量
    val offsetMap: Map[TopicPartition, Long] = OffsetManager.getOffset(topic,groupId)
    //  2 接收kafka数据
    var inputDstream: InputDStream[ConsumerRecord[String, String]]=null
    if(offsetMap!=null && offsetMap.size>0){
      inputDstream  = MyKafkaUtil.getKafkaStream(topic,ssc,offsetMap ,groupId)
    }else{
      inputDstream  = MyKafkaUtil.getKafkaStream(topic,ssc ,groupId)
    }

    //3  提取偏移量结束点
    var offsetRanges: Array[OffsetRange]=null  //driver
    val inputOffsetDstream: DStream[ConsumerRecord[String, String]] = inputDstream.transform { rdd =>
      offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges   // 每批次执行一次
      rdd
    }
    //4  转换结构 msg->   PageLog
    val pageLogDstream: DStream[PageLog] = inputOffsetDstream.map { record =>
      val jsonString: String = record.value()
      val pageLog: PageLog = JSON.parseObject(jsonString, classOf[PageLog])
      pageLog

    }
   // pageLogDstream.print(1000)



    // 5 筛选去重
    // 5.1 自我审查 ，凡是有last_page_id ，说明不是本次会话的第一个页面，直接过滤掉
    val filterBylastPageDstream: DStream[PageLog] = pageLogDstream.filter(pagelog =>
      if (pagelog.last_page_id == null) {
        true
      } else {
        false
      }
    )


    // 5.2 第三方审查 ，所有会话的第一个页面 去redis检查是否是今天的第一次
    val pageLogfilteredDstream: DStream[PageLog] = filterBylastPageDstream.mapPartitions { pageLogItr =>
      //查询redis 该用户是否访问过
      //type ? string list set zset hash   set    key?  dau:日期  field/score? 无  value?   uid(mid)
      // 写入api   ?  sadd     读取api?  sadd     过期时间？有  1天        //会有问题 zset
          val jedisClient: Jedis = RedisUtil.getJedisClient    //每批次每分区只使用一个连接
          val filteredList: ListBuffer[PageLog] = ListBuffer[PageLog]()
         val pageLoglist: List[PageLog] = pageLogItr.toList   //如果直接使用迭代器 ，那么size之后 ，迭代器无法继续使用
          println("redis过滤前： "+pageLoglist.size)
          for (pageLog <- pageLoglist ) {
            val dateFormat = new SimpleDateFormat("yyyy-MM-dd")
            val dateStr: String = dateFormat.format(new Date(pageLog.ts))
            val dauKey: String = s"DAU:${dateStr}"

            val isNewDau: lang.Long = jedisClient.sadd(dauKey, pageLog.mid)
            jedisClient.expire(dauKey,24*3600)//1日过期

            if (isNewDau == 1L) { //1 保存成功表示 是今日首次访问  保留   0 表示 之前已有访问 过滤掉
              filteredList.append(pageLog)
            }
          }
      println("redis过滤后： "+filteredList.size)
      jedisClient.close()
      filteredList.toIterator
    }

   // pageLogfilteredDstream.print(100)



    // 6 合并维度
    // 查redis 建连接 以分区为单位
    val dauInfoDstream: DStream[DauInfo] = pageLogfilteredDstream.mapPartitions { pageLogItr =>
      val jedisClient: Jedis = RedisUtil.getJedisClient
      val dauList: ListBuffer[DauInfo] = ListBuffer[DauInfo]()
      for (pageLog <- pageLogItr) {
        // 用户信息 关联
        val dimUserKey = s"DIM:USER_INFO:${pageLog.user_id}"
        val userInfoJson: String = jedisClient.get(dimUserKey)
        val userJsonObj: JSONObject = JSON.parseObject(userInfoJson)
        val birthday: String = userJsonObj.getString("birthday")
        val gender: String = userJsonObj.getString("gender")

        val dauInfo = new DauInfo()
        MyBeanUtils.copyProperties(pageLog, dauInfo)
        dauInfo.user_gender=gender

        val simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd")
        //生日变年龄
        if (birthday != null) {
//润年有误差
//          val bithdayDate: Date = simpleDateFormat.parse(birthday)
//          val diffTs: Long = new Date().getTime - bithdayDate.getTime
//          val age: Long = diffTs / 1000 / 3600 / 24 / 365

          //润年无误差
          val birthdayDate2: LocalDate = LocalDate.parse(birthday)
          val now: LocalDate = LocalDate.now()
          var period = Period.between(birthdayDate2, now);
          val years: Int = period.getYears
          dauInfo.user_age=years.toString




        }
        //地区维度查询 （作业， 需要使用新的数据生成器）
        val provinceKey = "DIM:BASE_PROVINCE:" + pageLog.province_id
        val provinceJson: String = jedisClient.get(provinceKey)
        if(provinceJson!=null&&provinceJson.size>0){
          val provinceJsonObj: JSONObject = JSON.parseObject(provinceJson)
          dauInfo.province_name = provinceJsonObj.getString("name")
          dauInfo.province_area_code = provinceJsonObj.getString("area_code") // ali datav  quickbi baidu suger
          dauInfo.province_3166_2 = provinceJsonObj.getString("iso_3166_2") // kibana
          dauInfo.province_iso_code = provinceJsonObj.getString("iso_code") // superset
        }

        //日期补充
        val dateFormat = new SimpleDateFormat("yyyy-MM-dd HH")
        val dtDate = new Date( dauInfo.ts)
        val dtHr: String = dateFormat.format(dtDate)
        val dtHrArr: Array[String] = dtHr.split(" ")
        dauInfo.dt = dtHrArr(0)
        dauInfo.hr  = dtHrArr(1)

        dauList.append(dauInfo)

      }

      jedisClient.close()
      dauList.toIterator
    }
    dauInfoDstream.cache()
    dauInfoDstream.print(1000)
    //  7写入es
   //精确一次消费 =  后置提交偏移+幂等
    dauInfoDstream.foreachRDD{rdd=>

      rdd.foreachPartition{dauInfoItr=>
           //因为是日活宽表 一天之内mid是不应该重复的
        val sourceList: List[(DauInfo, String)] = dauInfoItr.toList.map(dauInfo=>(dauInfo,dauInfo.mid) )
        if(sourceList.size>0){
          val dau: (DauInfo, String) = sourceList(0)

          //根据日期生成索引名
            val indexName=s"gmall0701_dau_info_${dau._1.dt}"
            MyEsUtil.bulkSave(sourceList,indexName)
        }

      }
     //driver  提交偏移量
      OffsetManager.saveOffset(topic,groupId,offsetRanges )
    }

    ssc.start()
    ssc.awaitTermination()

  }


  //破绽 来自于进程宕掉 过滤 写在redis了  数据写在es





  //状态还原
  // 在启动spark程序时执行
  // 该方法执行的位置  1 main下面 ？  2 transform里面 3 mapPartition里面 4 foreachRDD里
  def revertDauState(): Unit ={
  //  2  直接把es当日id号 全部取出 覆盖到redis中
  //  （一查 一写 ）
    // 一查  elaticsearch中的当日全部日活数据mid
      val esClient: RestHighLevelClient = MyEsUtil.getClient()
      //构建查询条件
      val dateFormat = new SimpleDateFormat("yyyy-MM-dd")
      val dt: String = dateFormat.format(new Date())
      val indexName=s"gmall0701_dau_info_$dt"
        val searchSourceBuilder = new SearchSourceBuilder()
       searchSourceBuilder.fetchSource("mid",null)
       searchSourceBuilder.size(1000)

       val searchRequest = new SearchRequest(indexName).source(searchSourceBuilder)
        var searchResponse:SearchResponse=null
       try{
            searchResponse = esClient.search(searchRequest,RequestOptions.DEFAULT)
       }catch {
         case exception: ElasticsearchStatusException=>{
             if(exception.status()==RestStatus.NOT_FOUND){
                println("恢复状态时："+exception.getIndex+"未找到，跳过")
                return
             }
         }
       }

       // 接收返回结果  返回结果放入 set
       val  dauSet=new util.HashSet[String]
       val hits: Array[SearchHit] = searchResponse.getHits.getHits
       for ( searchHits<- hits ) {
         val jsonSource: String = searchHits.getSourceAsString
         val jsonObj: JSONObject = JSON.parseObject(jsonSource)
         val mid: String = jsonObj.getString("mid")
         dauSet.add(mid)
       }
      println("当前es 的状态mid:"+dauSet)
      //  一些  把当日全部mid 写入redis
      val jedisClient: Jedis = RedisUtil.getJedisClient
      val dauKey: String = s"DAU:${dt}"
       jedisClient.del(dauKey)  //清空

       // 利用 pipeline 执行一个批量操作
       val pipeline: Pipeline = jedisClient.pipelined()
       import  scala.collection.JavaConverters._
       for ( mid <- dauSet.asScala ) {
         pipeline.sadd(dauKey,mid)
       }
       pipeline.sync()
       jedisClient.close()
       println("还原完成")


  }

}
